package net.fruit.mq.client.bean;

import java.lang.reflect.Method;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;

import net.fruit.mq.client.Consumer;
import net.fruit.mq.client.MQFactory;
import net.fruit.mq.client.MessageListener;
import net.fruit.mq.client.exception.LMQClientException;

public class ConsumerBean implements Consumer {
    private Properties properties;
    private Map<Subscription, MessageListener> subscriptionTable;

    private Consumer consumer;

    @Override
    public void start() {
        if (null == this.properties) {
            throw new LMQClientException("properties not set");
        }

        if (null == this.subscriptionTable) {
            throw new LMQClientException("subscriptionTable not set");
        }

        this.consumer = MQFactory.createConsumer(this.properties);

        Iterator<Entry<Subscription, MessageListener>> it = this.subscriptionTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<Subscription, MessageListener> next = it.next();
            this.subscribe(next.getKey().getTopic(), next.getKey().getExpression(), next.getValue());
        }

        this.consumer.start();
    }


    @Override
    public void shutdown() {
        if (this.consumer != null) {
            this.consumer.shutdown();
        }
    }


    @Override
    public void subscribe(String topic, String subExpression, MessageListener listener) {
        if (null == this.consumer) {
            throw new LMQClientException("subscribe must be called after consumerBean started");
        }
        this.consumer.subscribe(topic, subExpression, listener);
    }

    @Override
    public void unsubscribe(String topic) {
        if (null == this.consumer) {
            throw new LMQClientException("unsubscribe must be called after consumerBean started");
        }
        this.consumer.unsubscribe(topic);
    }

    public Properties getProperties() {
        return properties;
    }

    public void setProperties(Properties properties) {
        this.properties = properties;
    }

    public Map<Subscription, MessageListener> getSubscriptionTable() {
        return subscriptionTable;
    }

    public void setSubscriptionTable(Map<Subscription, MessageListener> subscriptionTable) {
        this.subscriptionTable = subscriptionTable;
    }

    @Override
    public boolean isStarted() {
        return this.consumer.isStarted();
    }

    @Override
    public boolean isClosed() {
        return this.consumer.isClosed();
    }
}
